// Copyright (C) 2010 and onwards Noam Shazeer and Georges Harik
// Pieces of this code may be licenced under the Apache License, version 2.0 
//   from Harik Shazeer Labs

#ifndef __SYNC_H_
#define __SYNC_H_

#include "util.h"

extern pthread_mutex_t print_lock;

struct SyncException {};

// const bool local_debug = false;

void * SyncMaintenanceLoop(void *sync);

struct Sync {
  // Many worker threads.  
  // One maintenance thread
  
  Sync() {
    needs_maintenance_ = false;
    num_workers_running_ = 0;
    next_worker_id_ = 0;
    pthread_mutex_init(&mutex_, NULL);
    pthread_cond_init(&maintenance_cond_, NULL);
    pthread_cond_init(&work_cond_, NULL);
    StartMaintenanceThread();
  }
  void AddMaintenanceRoutine(void * (*maintenance_routine)(void *)) {
    maintenance_routines_.push_back(maintenance_routine);
  }
  
  // Worker threads call this. 
  void StartWork();
  void StopWork();
  void CheckStop(); // may throw a SyncException
  void StopAll(); // call when maintenance is needed. 
    
  // Maintenance thread calls this:
  void StartMaintenance();
  void EndMaintenance();
  
  void MaintenanceLoop() {
    while (1) {
      StartMaintenance();
      for (uint64 i=0; i<maintenance_routines_.size(); i++) 
	(*(maintenance_routines_[i]))(NULL);
      EndMaintenance();
    }
  }
  void StartMaintenanceThread() {
    pthread_t * thread = new pthread_t;
    pthread_create(thread, NULL, &SyncMaintenanceLoop, (void *)this);
  }
  typedef void *(*fptr)(void *);
  vector<fptr> maintenance_routines_;
  
  struct Worker {
  Worker(Sync * sync) : sync_(sync), critical_(0) { 
    id_ = __sync_fetch_and_add(&sync->next_worker_id_, 1);
    sync_->StartWork(); 
  }
    virtual ~Worker() { 
      sync_->StopWork(); 
    }
    void CheckStop() { 
      if (sync_ && !critical_) sync_->CheckStop(); 
    }
    void StopAll() { sync_->StopAll(); CheckStop(); }
    Sync * sync_;
    uint64 critical_;
    int64 id_;
  };
  struct Critical {
    Critical(Worker *worker) {
      worker_ = worker;
      worker_->critical_++;
    } 
    ~Critical() { worker_->critical_--;}
    Worker *worker_;
  };
  
  bool needs_maintenance_;
  uint64 num_workers_running_;
  pthread_mutex_t mutex_;
  pthread_cond_t maintenance_cond_, work_cond_;
  int64 next_worker_id_;
};

struct RWLock {
  // holds a lock
  struct Reader {
    Reader(RWLock * rw) {
      // if (local_debug) cout << "::Reader " << (void*) this << endl;
      reading_ = writing_ = 0;
      pthread_mutex_init(&lock_, NULL);
      pthread_cond_init(&writer_done_cond_, NULL);
      rw_ = rw;
      {
	MutexLocker l(&rw_->lock_);
	rw_->readers_.push_back(this);
      }
      // pthread_mutex_lock(&lock_);
    }
    ~Reader() { 
      // if (local_debug) cout << "::~Reader " << (void*) this << endl;
      {
	MutexLocker l(&rw_->lock_);
	rw_->readers_.erase(find(rw_->readers_.begin(), 
				 rw_->readers_.end(), this));
      }
    }
    bool WriterWaiting() { return rw_->writer_waiting_; }
    void StartRead() { 
      if (reading_ + writing_ == 0) {
	// if (local_debug) cout << "::Reader::StartRead " << (void*) this << endl;
	pthread_mutex_lock(&lock_);
	while (rw_->writer_waiting_) {
	  pthread_cond_wait(&writer_done_cond_, &lock_);
	}
      }
      reading_++;
    }
    void EndRead() { 
      //if (local_debug) cout << "::Reader::EndRead " << (void*) this << endl;
      reading_--;
      if (reading_ + writing_ == 0) pthread_mutex_unlock(&lock_); 
    }
    void StartWrite() {
      if (writing_ == 0) {
	pthread_mutex_lock(&rw_->lock_);
	if (reading_) pthread_mutex_unlock(&lock_);
	rw_->writer_waiting_ = true;
	for (uint64 i=0; i<rw_->readers_.size(); i++)	  
	  pthread_mutex_lock(&(rw_->readers_[i]->lock_));
      }
      writing_++;
    }
    void EndWrite() {
      writing_--;
      if (writing_ == 0) {
	rw_->writer_waiting_ = false;
	for (uint64 i=0; i<rw_->readers_.size(); i++) {
	  pthread_cond_signal(&(rw_->readers_[i]->writer_done_cond_));
	  pthread_mutex_unlock(&(rw_->readers_[i]->lock_));
	}
	if (reading_) pthread_mutex_lock(&lock_);
	pthread_mutex_unlock(&rw_->lock_);	
      }
    }
    
    RWLock * rw_;
    pthread_cond_t writer_done_cond_;  
    pthread_mutex_t lock_;

    uint64 reading_;
    uint64 writing_;
  };

  RWLock() {
    //if (local_debug) cout << "::RWLock " << (void*) this << endl;
    pthread_mutex_init(&lock_, NULL); 
    writer_waiting_ = false;
  }

  struct ReadLocker {
    ReadLocker(Reader *reader) {
      reader_ = reader;
      reader_->StartRead();
    }
    ~ReadLocker() { reader_->EndRead(); }
    Reader *reader_;
  };
  struct WriteLocker {
    WriteLocker(Reader *reader) {
      reader_ = reader;
      reader_->StartWrite();
    }
    ~WriteLocker() { reader_->EndWrite(); } 
    Reader *reader_;
  };
  
  pthread_mutex_t lock_;
  bool writer_waiting_;
  vector<Reader *> readers_;
};

int test_rw_lock_main();

#endif
